package com.wei.inputformat;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class SmallFilesToSequenceFileConverter extends Configured implements Tool{
	
	static class SmallFilesMap extends Mapper<NullWritable, BytesWritable, Text, BytesWritable>{

		private Text filename=new Text();
		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			InputSplit inputSplit = context.getInputSplit();
			FileSplit fileSplit=(FileSplit) inputSplit;
			String name = fileSplit.getPath().getName();
			filename.set(name);
		}
		@Override
		protected void map(NullWritable key, BytesWritable value,
				Context context) throws IOException, InterruptedException {
			context.write(filename, value);
		}
		
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		System.setProperty("HADOOP_USER_NAME", "weicx");
		Job job = Job.getInstance(configuration,"small file combine");
		job.setMapperClass(SmallFilesMap.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(BytesWritable.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(BytesWritable.class);
		job.setInputFormatClass(WholeFileInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		boolean waitForCompletion = job.waitForCompletion(true);
		return waitForCompletion?0:1;
	}
	
	public static void main(String[] args) {
		try {
			int run = ToolRunner.run(new SmallFilesToSequenceFileConverter(),args);
			System.exit(run);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
